package reversion

import (
	"encoding/json"
	"fmt"
	"historyOfWikipedia/internal/compress"
	"historyOfWikipedia/internal/config"
	"historyOfWikipedia/internal/database/mongodb"
	"historyOfWikipedia/internal/model"
	"historyOfWikipedia/pb/reversion_pb"
	"io/ioutil"
	"net/http"
	"strconv"
	"time"

	log "github.com/sirupsen/logrus"
	"github.com/streadway/amqp"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var (
	inDetailQname  = "reversionDetailIn"
	outDetailQname = "reversionDetailOut"
)

func warnErr(err error) {
	if err != nil {
		log.Warn(err)
	}
}

func initDetailQueue() *amqp.Channel {
	connection, err := amqp.Dial(config.Env.AmqpUri)
	failOnError(err, "Failed to connect to RabbitMQ")
	channel, err := connection.Channel()
	failOnError(err, "Failed to connect to RabbitMQ")
	_, err = channel.QueueDeclare(
		inDetailQname, // name
		false,         // durable
		false,         // delete when unused
		false,         // exclusive
		false,         // no-wait
		nil,           // arguments
	)
	failOnError(err, "Failed to connect to RabbitMQ")
	_, err = channel.QueueDeclare(
		outDetailQname, // name
		false,          // durable
		false,          // delete when unused
		false,          // exclusive
		false,          // no-wait
		nil,            // arguments
	)
	failOnError(err, "Failed to connect to RabbitMQ")
	return channel
}

//
func ProducerDetail() {

	mongoObj := mongodb.NewMongoClient(config.Env.MongoUri)

	// 更新坐标
	mongoObj.Generate_Snapshot_Article()

	// 未完成的任务上传消息队列
	outchan, wg := mongoObj.Get_Unfinish_Revision_detail()

	channel := initDetailQueue()
	// 发送任务
	var ct int
	for item := range outchan {
		ct += 1
		title, err := mongoObj.Get_article_title(item.ArticleID)
		if err != nil {
			log.Println("title query fail", err)
			continue
		}
		err = channel.Publish(
			"",            // exchange
			inDetailQname, // routing key
			false,         // mandatory
			false,         // immediate
			amqp.Publishing{
				ContentType: "application/octet-stream",
				Body:        []byte{},
				Headers: amqp.Table{
					"ID_Hex":    item.ID.Hex(),
					"Title":     title,
					"ArticleID": item.ArticleID,
					"Revid":     item.Revid,
				},
			})
		if err != nil {
			log.Println(err)
		}
	}
	wg.Wait()
	channel.Close()
	log.Info("上传队列结束，共上传任务数", ct)
}

func WorkerDetail() {
	channel := initDetailQueue()
	err := channel.Qos(
		10,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")

	msgs, err := channel.Consume(
		inDetailQname, // queue
		"",            // consumer
		false,         // auto-ack
		false,         // exclusive
		false,         // no-local
		false,         // no-wait
		nil,           // args
	)
	failOnError(err, "Failed to register a consumer")

	if config.Env.HttpProxy != "" {
		log.Info("use proxy:", config.Env.HttpProxy)
		var PTransport = &http.Transport{Proxy: http.ProxyFromEnvironment}
		client = http.Client{Transport: PTransport, Timeout: 120 * time.Second}
	} else {
		log.Info("no proxy")
		client = http.Client{Timeout: 120 * time.Second}
	}

	for d := range msgs {

		ID_Hex := d.Headers["ID_Hex"].(string)
		Title := d.Headers["Title"].(string)
		ArticleID := d.Headers["ArticleID"].(int64)
		Revid := d.Headers["Revid"].(int64)

		var ret_data reversion_pb.Reversion_Detail_Job

		state, revs, err := get_reversions_content(Title, Revid)
		if err != nil {
			log.Info("请求错误:", Title, err)
			ret_data.State = false
			ret_data.Message = err.Error()
		} else if state == 0 {
			log.Info("没有找到该条记录:", Title)
			ret_data.State = false
			ret_data.Message = "没有找到该条记录"
		} else if state == 1 {
			log.Println("查询完毕:", Title)
			ret_data.State = true
		}
		// 正确的情况，保存 revdetail
		if revs != nil {
			if Revid != revs.RevID {
				ret_data.State = false
				log.Infof("revid 不匹配,old revid:%d, new revid:%d", Revid, revs.RevID)
				ret_data.Message = fmt.Sprintf("revid 不匹配,old revid:%d, new revid:%d", Revid, revs.RevID)
			}
			ret_data.ReversionDetail = revs
		}

		go func(qd amqp.Delivery) {
			log.Println("over revid:", Revid)
			data, err := proto.Marshal(&ret_data)
			failOnError(err, "Marshal 失败")
			compressBody, err := compress.CompressData(data)
			failOnError(err, "compress 失败")
			for {
				err := channel.Publish(
					"",             // exchange
					outDetailQname, // routing key
					false,          // mandatory
					false,          // immediate
					amqp.Publishing{
						ContentType: "application/octet-stream",
						Body:        compressBody,
						Headers: amqp.Table{
							"ID_Hex":    ID_Hex,
							"Title":     Title,
							"ArticleID": ArticleID,
							"Revid":     Revid,
						},
					})
				if err != nil {
					log.Println("upload", err, " wait 1s ")
					<-time.After(1 * time.Second)
					continue
				}
				break
			}
			qd.Ack(false)
		}(d)
	}

	forever := make(chan bool)
	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

// state 0, fail, 1 success
// lastrevid
// revionse
func get_reversions_content(article_name string, rvstartid int64) (state int, revs *reversion_pb.Reversion_Detail, err error) {
	// log.Println("开始任务:", article_name, rvendid)
	article_name = replace_wiki_name(article_name)

	base_url := "https://en.wikipedia.org/w/api.php"
	req, _ := http.NewRequest("GET", base_url, nil)
	q := req.URL.Query()
	q.Add("action", "query")
	q.Add("format", "json")
	q.Add("prop", "revisions")
	q.Add("rvlimit", "1")
	q.Add("rvprop", "timestamp|ids|content")
	q.Add("rvslots", "*")
	q.Add("titles", replace_wiki_name(article_name))
	q.Add("rvstartid", strconv.Itoa(int(rvstartid)))
	req.URL.RawQuery = q.Encode()

	var resp *http.Response

	log.Info(req.URL.String())
	for i := 0; i <= 3; i++ {
		resp, err = client.Do(req)
		if err != nil {
			if i == 3 {
				return
			}
			log.Println("请求失败，正在重试：", i, err)
			time.Sleep(1 * time.Second)
			continue
		}
		if resp.StatusCode < 200 || resp.StatusCode >= 300 {
			time.Sleep(1 * time.Second)
			continue
		}
		break
	}

	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		err = fmt.Errorf("statu no 200: %d", resp.StatusCode)
		time.Sleep(1 * time.Second)
		return
	}
	content, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	state, revs = formatDetailResponse(content)
	return
}

// stat
func formatDetailResponse(responseBody []byte) (state int, rev *reversion_pb.Reversion_Detail) {
	// new object
	rev = &reversion_pb.Reversion_Detail{}

	var repData map[string]interface{}
	json.Unmarshal(responseBody, &repData)

	// 没有query的情况
	if _, found := repData["query"].(map[string]interface{})["pages"]; !found {
		log.Info("no page found")
		return
	}
	pages := repData["query"].(map[string]interface{})["pages"].(map[string]interface{})
	// 当前的 page
	var mypage map[string]interface{}
	for key, page := range pages {
		if key == "-1" {
			log.Info("key == -1")
			return
		}
		mypage = page.(map[string]interface{})
		break
	}

	// 补丁
	if mypage["revisions"] == nil {
		return
	} else if mypage["missing"] != nil && mypage["revisions"] == nil {
		return
	}
	rev.ArticleID = int64(mypage["pageid"].(float64))

	revisions := mypage["revisions"].([]interface{})
	if len(revisions) != 1 {
		log.Info("revisions 长度部位 1：", len(revisions))
		return
	}
	item := revisions[0].(map[string]interface{})
	rev.RevID = int64(item["revid"].(float64))

	// 分步解析，应对中间情况
	soltMain := item["slots"].(map[string]interface{})["main"].(map[string]interface{})
	if _, stats := soltMain["texthidden"]; stats {
		log.Infof("text hidden,revID: %d", rev.RevID)
		rev.Texthidden = true
	}else if _, stats := soltMain["textmissing"]; stats {
		log.Infof("text hidden,revID: %d", rev.RevID)
		rev.Texthidden = true
	} else {
		rev.Text = soltMain["*"].(string)
	}
	revTimestamp, _ := time.Parse("2006-01-02T15:04:05Z", item["timestamp"].(string))
	rev.Timestamp = timestamppb.New(revTimestamp)
	state = 1
	return
}

func ConsumerDetail() {

	mongoObj := mongodb.NewMongoClient(config.Env.MongoUri)
	channel := initDetailQueue()
	err := channel.Qos(
		10,    // prefetch count
		0,     // prefetch size
		false, // global
	)
	failOnError(err, "Failed to set QoS")
	msgs, err := channel.Consume(
		outDetailQname, // queue
		"",             // consumer
		false,          // auto-ack
		false,          // exclusive
		false,          // no-local
		false,          // no-wait
		nil,            // args
	)
	failOnError(err, "Failed to register a consumer")

	for d := range msgs {

		ArticleID := d.Headers["ArticleID"].(int64)
		Revid := d.Headers["Revid"].(int64)
		ID_Hex := d.Headers["ID_Hex"].(string)
		ID, err := primitive.ObjectIDFromHex(ID_Hex)
		if err != nil {
			log.Warn("hex 转换 object id fail")
			d.Ack(true)
		}
		log.Infof("start ID :%d", Revid)
		data, err := compress.DecompressData(d.Body)
		if err != nil {
			log.Info("解压缩失败，跳过 article:", ID)
			err = d.Ack(true)
			warnErr(err)
			continue
		}
		var reversion_detail_job_obj reversion_pb.Reversion_Detail_Job
		err = proto.Unmarshal(data, &reversion_detail_job_obj)
		if err != nil {
			log.Info("Unmarshal 失败，跳过 article:", ID)
			continue
		}

		if !reversion_detail_job_obj.State {
			// 采集失败的处理方法
			err := mongoObj.Mark_update_revision_detail_fail(ID, reversion_detail_job_obj.Message)
			warnErr(err)
			log.Info("ID fail")
		} else {
			// 成功就选择更新数据
			reversionDetail := model.RevisionDetailObj{
				ID:        ID,
				Revid:     Revid,
				ArticleID: ArticleID,
				Timestamp: reversion_detail_job_obj.ReversionDetail.Timestamp.AsTime(),
				Text:      reversion_detail_job_obj.ReversionDetail.Text,
			}
			if reversion_detail_job_obj.ReversionDetail.Texthidden {
				// texhidden ，需要标记
				reversionDetail.TextHidden = true
				// 标记 text hidden
				mongoObj.Mark_revision_text_hidden(ID)
			}

			// log.Info("回收数据：", reversionDetail)
			// 更新到数据库
			_, err := mongoObj.Update_Revision_Detail(&reversionDetail, ID)
			if err == nil {
				mongoObj.Mark_update_revision_detail_finish(ID)
			} else {
				log.Warn("最后一步，更新失败，", err)
			}
		}
		err = d.Ack(true)
		warnErr(err)
	}

	// forever := make(chan bool)
	// log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	// <-forever
}
